/**
  * @file   MzSequenceFileCaidRecordReader.java
  * @author Jiang Longlong <jianglonglong@miaozhen.com>
  * @date   2014-11-17
  *
  * @brief
  *
  *
  */

package com.supertool.tong.merger.util;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import java.io.IOException;
import java.util.*;
import java.io.BufferedReader;
import java.io.FileReader;

public class MzSequenceFileCaidRecordReader extends RecordReader<LongWritable, Text> {
	
    
    //protected Configuration conf;
	private LongWritable key = new LongWritable(0);
	private Text value = new Text();
	private List<String> caids;
	private int i=0;
	private final String CAIDMETA="ca_";
	private final  String MSIDMETA="ms_";
	private MzSequenceFile.Reader in;
	private Path path;
	private FileSystem fs ;
	private Configuration conf;
	private long start=0L;
	private long end=0L;
	private int type;//0 for total ,1 caid,2 msid,3 begincaid-endcaid 
	
	public MzSequenceFileCaidRecordReader(Configuration conf, FileSplit split)
            throws IOException 
	{
		this.conf=conf;
		path = split.getPath();
		fs = path.getFileSystem(conf);
		i = 0;
		caids=new ArrayList<String>();
		String filename=conf.get("caidfilename");
		String filename2=conf.get("msidfilename");
		String begcaid=conf.get("begincaid");
		String endcaid=conf.get("endcaid");
		if(filename!=null)
		{
			FileReader fr = new FileReader(filename); 
			BufferedReader reader = new BufferedReader(fr);
			String id = null;
			while((id = reader.readLine()) != null)
				caids.add(CAIDMETA+id);
			type=1;
		}
		else if(filename2!=null)
		{
			FileReader fr = new FileReader(filename2); 
			BufferedReader reader = new BufferedReader(fr);
			String id = null;
			while((id = reader.readLine()) != null)
				caids.add(MSIDMETA+id);
			type=2;
		}
		else if(begcaid!=null&&endcaid!=null)
		{
			int begint = Integer.parseInt(begcaid);
			int endint = Integer.parseInt(endcaid);
			MzSequenceFile.Reader seqReader = new MzSequenceFile.Reader(fs, path, fs.getFileStatus(path).getLen(), conf, null);
			MzSequenceFile.MetaData metadata=seqReader.getMetadata();
			TreeMap<String, MzSequenceFile.SMeta>  smetas=metadata.getMetadata();
			Iterator titer=smetas.entrySet().iterator();  
			while(titer.hasNext()){  
				Map.Entry ent=(Map.Entry )titer.next();  
				String keyt=ent.getKey().toString().substring(3);
				try
				{
					int i = Integer.parseInt(keyt);
					if(i>=begint&&i<=endint)
					caids.add(CAIDMETA+keyt);		
				}
				catch(Exception e){}
				finally{}
			}
			type=3;
		}  
		else 
		{
			String meta=null;
			caids.add(meta);
			type=0;
		}
		if(caids.size()> 0)
		{
			in = new MzSequenceFile.Reader(fs, path, fs.getFileStatus(path).getLen(), conf, caids.get(i));
			if(type==0)
			{
				this.start=this.in.getStart();
				this.end=this.in.getEnd();
			}
		}else if(caids.size() == 0){
			in = new MzSequenceFile.Reader(fs, path, fs.getFileStatus(path).getLen(), conf, null);
                        if(type==1)
                        {
                                this.start=this.in.getStart();
                                this.end=this.in.getEnd();
                        }
		}
	}
   
	public void initialize(InputSplit split, TaskAttemptContext context)
            throws IOException, InterruptedException {}
	public Class getValueClass() 
	{
		 return in.getValueClass(); 
	}
	public LongWritable getCurrentKey() 
	{
        	return this.key;
    	}
	public Text getCurrentValue() 
	{
        	return this.value;
    	}
	public  boolean nextKeyValue() throws IOException 
	{
		if(in!=null)
		{
			if(in.next((Object)this.value))
				return true;
			else
			{
				i++;
				while(caids.size()>i)
				{
					in = new MzSequenceFile.Reader(fs, path, fs.getFileStatus(path).getLen(), conf, caids.get(i));
					if(in.next((Object)this.value))
						return true;
					else
						i++;
				}
			}	
					
		}
		else
			return false;
		return false;
	}
	public float getProgress() throws IOException 
	{
		if(type==0)
        	{
			if (end == start) {
				return 0.0f;
			} 
			else 
			{
				return Math.min(1.0f, (in.getPos() - start) / (float)(end - start));
			}
		}
		else
			return  i/(float)caids.size();
	}  
	public long getPos() throws IOException 
	{
        	if(in!=null)
			return in.getPos();
		else
			return 0;
	}

	protected void seek(long pos) throws IOException {
        	throw new IOException("MzSequenceFile does not support seek");
    	}
	public void close() throws IOException {  }

}
